﻿using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Threading;

namespace MqResearchCSharp.ActiveMq.ActiveMQ
{
    /// <summary>
    /// 异步执行任务时候，用于同步计算结果
    /// </summary>
    public class TaskInfo
    {
        public TaskMessage TaskMessage { get; set; }
        public AutoResetEvent Event { get; set; }
        public TaskInfo(TaskMessage taskMessage)
        {
            TaskMessage = taskMessage;
            Event = new AutoResetEvent(false);
        }
    }

    public class ActiveMQ
    {
        public string ClientId = "";
        public ITemporaryQueue TemporaryQueue;

        private IConnectionFactory _factory;
        private IConnection _connection;
        private ISession _session;
        private IMessageProducer _producer;
        private IMessageConsumer _consumer;
        private ITextMessage _textMessage;

        private OtcLog _log = new OtcLog(typeof(ActiveMQ));
        private ConcurrentDictionary<string, TaskInfo> _taskMap = new ConcurrentDictionary<string, TaskInfo>();

        public ActiveMQ(string clientId = "Server", bool isLocalMachine = true, string remoteAddress = "")
        {
            try
            {
                if (isLocalMachine)
                {
                    _factory = new ConnectionFactory("tcp://127.0.0.1:61616/");
                }
                else
                {
                    _factory = new ConnectionFactory(remoteAddress); //写tcp://192.168.1.111:61616的形式连接其他服务器上的ActiveMQ服务器           
                }
                ClientId = clientId;
                //(_factory as ConnectionFactory).AsyncSend = true;
                _log.BindHead($"[OtcActiveMQ],[{ClientId}]");
                _connection = _factory.CreateConnection();
                _connection.ClientId = clientId;
                _connection.Start();
                _session = _connection.CreateSession();
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                throw e;
            }
            _log.Debug("Begin connection...");
        }

        /// <summary>
        /// 同步初始化
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool InitServer(string queueName)
        {
            try
            {
                _producer = _session.CreateProducer();
                _consumer = _session.CreateConsumer(new ActiveMQQueue(queueName));
                _consumer.Listener += _consumer_Listener;

                _textMessage = _session.CreateTextMessage();
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        /// <summary>
        /// 同步初始化
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        public bool InitClient(string queueName)
        {
            try
            {
                TemporaryQueue = _session.CreateTemporaryQueue();
                _producer = _session.CreateProducer(new ActiveMQQueue(queueName));
                _consumer = _session.CreateConsumer(TemporaryQueue);
                _consumer.Listener += _consumer_Listener;

                _textMessage = _session.CreateTextMessage();
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        private void _consumer_Listener(IMessage message)
        {
            var revMessage = message as ITextMessage;
            TaskMessage taskMsg = JsonConvert.DeserializeObject<TaskMessage>(revMessage.Text);
            if (_taskMap.ContainsKey(taskMsg.TaskID))
            {
                _taskMap[taskMsg.TaskID].TaskMessage = taskMsg;
                _taskMap[taskMsg.TaskID].Event.Set();
                _log.Debug($"[{taskMsg.TaskName}],[{taskMsg.TaskID}],任务执行完毕,耗时:{taskMsg.TotalSeconds}秒,错误信息:{taskMsg.Error}");
            }
            else
            {
                _log.Debug($"[{taskMsg.TaskName}],[{taskMsg.TaskID}],任务未找到...");
            }
        }

        public bool InitServer(string name, MessageListener action)
        {
            try
            {
                _producer = _session.CreateProducer();
                _consumer = _session.CreateConsumer(new ActiveMQQueue(name));
                _consumer.Listener += action;

                _textMessage = _session.CreateTextMessage();
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        public bool InitClient(string name, MessageListener action)
        {
            try
            {
                TemporaryQueue = _session.CreateTemporaryQueue();
                _producer = _session.CreateProducer(new ActiveMQQueue(name));
                _consumer = _session.CreateConsumer(TemporaryQueue);
                _consumer.Listener += action;

                _textMessage = _session.CreateTextMessage();
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        public bool ReplayMessage(IDestination destination, TaskMessage taskMessage, MsgDeliveryMode deliveryMode = MsgDeliveryMode.Persistent, MsgPriority priority = MsgPriority.Normal)
        {
            if (_producer == null)
            {
                _log.Error("Call InitQueueOrTopic() first!!");
                return false;
            }

            _log.Debug($"Replay:{taskMessage.TaskName},{taskMessage.TaskID}");

            try
            {
                var objectMessage = _session.CreateObjectMessage(taskMessage);
                objectMessage.NMSCorrelationID = taskMessage.TaskID;
                objectMessage.Body = taskMessage;

                _producer.Send(destination, objectMessage, deliveryMode, priority, TimeSpan.MinValue);
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        public bool AsyncRunTask(TaskMessage taskMessage, MsgDeliveryMode deliveryMode, MsgPriority priority)
        {
            if (_producer == null)
            {
                _log.Error("Call InitQueueOrTopic() first!!");
                return false;
            }

            _log.Debug($"Send:{taskMessage.TaskName},{taskMessage.TaskID}");

            try
            {
                var objectMessage = _session.CreateObjectMessage(taskMessage);
                objectMessage.NMSCorrelationID = taskMessage.TaskID;
                objectMessage.NMSReplyTo = TemporaryQueue;

                _producer.Send(objectMessage, deliveryMode, priority, TimeSpan.MinValue);
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return false;
            }

            return true;
        }

        public TaskMessage RunTask(TaskMessage taskMessage, MsgDeliveryMode deliveryMode, MsgPriority priority)
        {
            if (_producer == null)
            {
                _log.Error("Call InitQueueOrTopic() first!!");
                return null;
            }

            _log.Debug($"Send:{taskMessage.TaskName},{taskMessage.TaskID}");

            var taskInfo = new TaskInfo(taskMessage);
            try
            {
                var objectMessage = _session.CreateTextMessage(JsonConvert.SerializeObject(taskMessage));
                objectMessage.NMSCorrelationID = taskMessage.TaskID;
                objectMessage.NMSReplyTo = TemporaryQueue;

                _taskMap.TryAdd(taskMessage.TaskID, taskInfo);
                _producer.Send(objectMessage, deliveryMode, priority, TimeSpan.MinValue);
            }
            catch (Exception e)
            {
                _log.Error($"{e.Message}");
                return null;
            }

            if (taskInfo.Event.WaitOne(taskInfo.TaskMessage.TimeOut))
            {
                return taskInfo.TaskMessage;
            }
            else
            {
                _log.Error($"[{taskMessage.TaskName}],[{taskMessage.TaskID}],任务执行超时......");
                return null;
            }
        }

        public void ShutDown()
        {
            _log.Debug("Close connection and session...");
            _session.Close();
            _connection.Close();
        }

        public bool AsyncRunTask(TaskMessage task)
        {
            return AsyncRunTask(task, MsgDeliveryMode.NonPersistent, MsgPriority.Normal);
        }

        public TaskMessage RunTask(TaskMessage task)
        {
            return RunTask(task, MsgDeliveryMode.NonPersistent, MsgPriority.Normal);
        }

        public void Close()
        {
            if (_connection != null)
            {
                _connection.Close();
            }
        }
    }
}